package com.distributed.master;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class MasterServer extends Thread {
    // server socket
    private final ServerSocket server_socket;

    public static HashMap<String, List<String>> directory = new HashMap<>(); // region_ip : tables
    public static HashMap<String, String> region_names = new HashMap<>();    // region_id : region_name
    public static int regions_num = 0;

    public static final int MASTER_PORT = 8000;
    public static final int REGION_LISTEN_PORT = 8002;

    // public static int num_of_region_servers = 0;

    /*
     * Author: Xinbei Wu
     * Name: MasterServer
     * Function: Create and Init MasterServer(server_socket)
     * Params: port(int | port number)
     * Return: void
     *  */
    public MasterServer(int port) throws IOException {
        server_socket = new ServerSocket(port);
        server_socket.setSoTimeout(Integer.MAX_VALUE);
    }

    /*
     * Author: Xinbei Wu
     * Name: run()
     * Function: run master server
     * Params: void
     * Return: void
     *  */
    public void run(){
        while (true) {
            try {
                // get socket
                Socket server = server_socket.accept();
                // get input stream
                DataInputStream input_stream = new DataInputStream(server.getInputStream());

                String response;
                String url;
                String[] table_name = input_stream.readUTF().split(" ");

                // parse table name
                if (table_name.length == 1) { // show/drop table
                    url = getUrlByTableName(table_name[0]);
                    if (url.equals("-1")) {
                        response = "[WARN] The table does not exist.";
                    } else {
                        response = url;
                    }
                } else { // create table
                    url = getUrlByTableName(table_name[1]);
                    if (url.equals("-1")) {
                        response = getMostFreeRegionServer();
                    } else {
                        response = "[WARN] The table already exists.";
                    }
                }

                // output
                DataOutputStream output_stream = new DataOutputStream(server.getOutputStream());
                output_stream.writeUTF(response);

                // close
                server.close();
            } catch(SocketTimeoutException s) {
                System.out.println("[ERROR] Socket times out.");
                break;
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /*
     * Author: Xinbei Wu
     * Name: getUrlByTableName
     * Function: find region url stores certain table
     * Params: table_name(String | table name)
     * Return: region_url(String | region url)
     *  */
    public String getUrlByTableName(String table_name){
        for (HashMap.Entry<String, List<String>> entry : directory.entrySet()) {
            if (entry.getValue().contains(table_name)) {
                return new String(entry.getKey());
            }
        }
        return new String("-1");
    }

    /*
     * Author: Xinbei Wu
     * Name: getMostFreeRegionServer
     * Function: find most free region server
     * Params: void
     * Return: region_url(String | region url)
     *  */
    public String getMostFreeRegionServer(){
        int min = Integer.MAX_VALUE;
        String free_region = "";
        for (Map.Entry<String, List<String>> entry : directory.entrySet()) {
            if (entry.getValue().size() < min) {
                min = entry.getValue().size();
                free_region = entry.getKey();
            }
        }
        return free_region;
    }

    public static void main(String[] args) throws Exception {
        Thread region_thread = new RegionServerConnector(8002);
        region_thread.start();

        Thread master_thread = new MasterServer(8000);
        master_thread.start();
        Thread zookeeper_thread = new ZookeeperConnector(directory);
        zookeeper_thread.start();
    }

}
